package com.syf.kit.soa.core.send.protocol.handle;

import com.syf.kit.soa.core.SyfRpcMsg;
import com.syf.kit.soa.core.SyfSOAProtocolCont;
import com.syf.kit.soa.core.send.protocol.BeatRpcMsgBuilder;
import com.syf.kit.soa.core.send.protocol.TcpResponseCoordinator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RpcKeepAliveSendHandler extends SimpleChannelInboundHandler<SyfRpcMsg> {
    /**
     * 魔数
     */
    private String magicNum;

    /**
     * 明确连接是否可用的共享参数key
     */
    private AttributeKey<Boolean> USABLE_ATTR_KEY;

    private CountDownLatch beatLatch;

    private ExecutorService beatExecutor = Executors.newSingleThreadExecutor();

    public RpcKeepAliveSendHandler(String magicNum, AttributeKey<Boolean> USABLE_ATTR_KEY) {
        super();
        this.magicNum = magicNum;
        this.USABLE_ATTR_KEY = USABLE_ATTR_KEY;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SyfRpcMsg respMsg) throws Exception {
        if (respMsg.getReqType() == SyfSOAProtocolCont.BEAT_REQ_TYPE) {
            if (log.isTraceEnabled()) {
                log.trace("RPC-TCP-CONNECT:{},收到心跳!", ctx.channel().remoteAddress());
            }
            beatLatch.countDown();
            return;
        }

        TcpResponseCoordinator.getInstance().onResponse(respMsg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.channel().attr(USABLE_ATTR_KEY).set(true);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.channel().attr(USABLE_ATTR_KEY).set(false);
    }

    void sendPing(ChannelHandlerContext ctx) {
        beatLatch = new CountDownLatch(1);
        ctx.writeAndFlush(BeatRpcMsgBuilder.build(magicNum));
        if (log.isTraceEnabled()) {
            log.trace("RPC-TCP-CONNECT:{},发出心跳!", ctx.channel().remoteAddress());
        }

        beatExecutor.submit(()->{
            try {
                if (!beatLatch.await(2000, TimeUnit.MILLISECONDS)) {
                    // 心跳超时，说明线路已断
                    log.warn("警告：RPC-TCP-CONNECT:{},心跳超时！", ctx.channel().remoteAddress());
                    ctx.channel().attr(USABLE_ATTR_KEY).set(false);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            log.trace("get event : [{}]", event.state());
            if (event.state() == IdleState.ALL_IDLE) {
                // 空闲事件触发后，主动ping
                log.trace("send ping");
                sendPing(ctx);
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("远程通道: [{}], 出现异常:{}", ctx.channel().remoteAddress().toString(), cause.getMessage());
        log.error("通道出现异常", cause);
    }
}
